package com.ldn.server.mqtt.covert;

import com.ldn.common.basic.dispose.bo.MqttRocketmqBo;
import com.ldn.common.basic.enums.mqtt.MMessageTypeEnums;
import com.ldn.common.basic.utils.system.DateUtils;
import com.ldn.common.basic.utils.system.StrUtils;
import com.ldn.server.mqtt.config.CovertType;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.MqttPublishMessage;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Component;
import org.tio.core.ChannelContext;

import java.util.HashMap;
import java.util.Map;

/**
 * 会员用户端消息处理器
 *  <p>
 *      主要用于处理会员用户上线，离线，消息处理等不包括用户订阅
 *  </p>
 *
 * @author 李大牛家的牛 (scafel@163.com)
 * @product serverless
 * @className MemberCovert
 * @createDate 2025/2/25 - 13:46
 * @description 会员用户端消息处理器
 * @see CovertType
 **/
@Slf4j
@Component
public class MemberCovert implements CovertType {

    private static final String TAG = "MEMBER-";

    @Resource
    private StreamBridge streamBridge;

    // 维护一个本类的静态变量 方便使用相关注入属性
    private static MemberCovert covert;
    private String mpp001;

    @PostConstruct
    public void init() {
        covert = this;
    }

    @Override
    public boolean checkAuthenticate(String uniqueId, String userName, String password) {
        // 验证账号是否存在并验证状态是否正确
        // 发送认证消息
        return true;
    }

    @Override
    public void messageHandler(ChannelContext context, String clientId, String topic, MqttQoS qoS, MqttPublishMessage message) {
        // 记录原始消息
        // 将消息发送到消息队列让队列中的信息去处理
    }

    @Override
    public void online(ChannelContext context, String clientId, String username) {
        // 记录原始消息
        // 将消息发送到消息队列让队列中的信息去处理
    }

    @Override
    public void offline(ChannelContext context, String clientId, String username, String reason) {
        // 记录原始消息
        // 将消息发送到消息队列让队列中的信息去处理
    }

    @Override
    public boolean checkCanSubTopic(String clientId, String topicFilter) {
        return true;
    }

    @Override
    public void onSubscribed(ChannelContext context, String clientId, String topicFilter, int qos) {
        String code = StrUtils.sub(clientId, TAG.length(), clientId.length());
        // 将消息发送到消息队列让队列中的信息去处理
        MqttRocketmqBo bo = new MqttRocketmqBo();
        bo.setClientCode(code);
        bo.setMessageType(MMessageTypeEnums.SUBSCRIBE.getCode());
        bo.setCreateTime(DateUtils.getCurrentMS());
        bo.setTopic(topicFilter);
        bo.setQos(qos);
        covert.sendRocketmq(bo);
    }

    @Override
    public void onUnsubscribed(ChannelContext context, String clientId, String topicFilter) {
        String code = StrUtils.sub(clientId, TAG.length(), clientId.length());
        // 将消息发送到消息队列让队列中的信息去处理
        MqttRocketmqBo bo = new MqttRocketmqBo();
        bo.setClientCode(code);
        bo.setMessageType(MMessageTypeEnums.UNSUBSCRIBE.getCode());
        bo.setCreateTime(DateUtils.getCurrentMS());
        bo.setTopic(topicFilter);
        covert.sendRocketmq(bo);
    }

    @Override
    public boolean checkCanPublish(String clientId, String topicFilter, int value, boolean isRetain) {
        return true;
    }

    @Override
    public String getTag() {
        return TAG;
    }

    private void sendRocketmq(MqttRocketmqBo bo) {
        try {
            log.info("RocketMq消息包   RocketMqMessage：{}", bo);
            covert.streamBridge.send("mqttMember-out-0", bo);
        } catch (Exception e) {
            log.error("RocketMq消息包发送失败，原因：{}", e.getMessage());
        }
    }
}
